#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <signal.h>
#include <pthread.h>
#include <errno.h>
// #include "shmem.h"
#include <stdbool.h>
#include "MQTTClient.h"
#include "shmem.h"
#include "cJSON.h"

// ==================== 宏 ====================
#define N 32
#define SHM_SIZE 4096 // 共享内存大小
// MQTT连接参数
// #define ADDRESS "tcp://localhost:1883" // MQTT代理地址
#define CLIENTID "ExampleClientSub" // 客户端ID
#define PUB_TOPIC "up"              // 发布主题
#define SUB_TOPIC "down"            // 订阅主题
#define QOS 1                       // 服务质量等级
#define TIMEOUT 10000L              // 超时时间(毫秒)
// ==================== 定义 ====================
// mqtt服务器初始化配置
volatile MQTTClient_deliveryToken deliveredtoken;
MQTTClient client;
MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;
MQTTClient_message pubmsg = MQTTClient_message_initializer;
MQTTClient_deliveryToken token;
int rc;
int ch;
// 设备数据类型
typedef union
{
    int b_val;   // 布尔型
    int i_val;   // 整型
    float f_val; // 浮点型
} val_t;

// 共享内存节点结构
typedef struct
{
    int key;       // 节点唯一标识
    int type;      // 数据类型 (1:bool, 2:int, 3:float)
    int dev_type;  // 设备类型
    val_t old_val; // 旧值
    val_t new_val; // 新值
    int ret;       // 状态码
} std_node;

static struct shm_param para; // 共享内存所需
static int *total;            // 指向共享内存中数据节点总个数
static std_node *node_arr;    // 指向共享内存中节点缓存数组头
static int stmsize;           // 点表中stm32的节点数量

// 上报线程传递的参数
typedef struct
{
    pthread_t rpt_id; // 线程标识 (除刷新上报外，其他模式持续至程序结束)
    int tim;          // 周期上报的时间
} pth_report;

// 指令处理线程传递的参数
typedef struct
{
    char *msg;
    pthread_t msg_id;
} pth_msg_recv;

// ==================== 函数声明 ====================
void *handler_period(void *arg);
void *handler_change(void *arg);
void *handler_refresh(void *arg);
void delivered(void *context, MQTTClient_deliveryToken dt);
int msgarrvd(void *context, char *topicName, int topicLen, MQTTClient_message *message);
void connlost(void *context, char *cause);
// ==================== 主函数 ====================
int main(int argc, char const *argv[])
{
    // ---------- 1. 读取并转化配置点表 ----------

    FILE *file = fopen("./config/config.json", "r");
    if (file == NULL)
    {
        perror("fopen err\n");
        return -1;
    }
    char configbuf[N * N * 2] = {0};
    size_t len = fread(configbuf, sizeof(char), N * N * 2, file);
    if (len < 0)
    {
        printf("fread err\n");
        return -1;
    }
    // printf("%s\n", configbuf); // 查看点表能否完整读出
    // 把传入的字符串转成cJSON的结构
    cJSON *root = cJSON_Parse(configbuf);
    if (root == NULL)
    {
        printf("cJSON_Parse err\n");
        return -1;
    }

    // ---------- 2. 建立mqtt服务器连接 ----------

    // 解析mqtt服务器配置
    cJSON *mqtt_server = cJSON_GetObjectItem(root, "mqtt_server");
    cJSON *addr = cJSON_GetObjectItem(mqtt_server, "addr");
    cJSON *port = cJSON_GetObjectItem(mqtt_server, "port");
    char ADDRESS[N];
    snprintf(ADDRESS, N, "tcp://%s:%d", addr->valuestring, port->valueint);

    // 建立mqtt服务器连接

    MQTTClient_create(&client, ADDRESS, CLIENTID,
                      MQTTCLIENT_PERSISTENCE_NONE, NULL);
    conn_opts.keepAliveInterval = 20;
    conn_opts.cleansession = 1;

    MQTTClient_setCallbacks(client, NULL, connlost, msgarrvd, delivered);

    if ((rc = MQTTClient_connect(client, &conn_opts)) != MQTTCLIENT_SUCCESS)
    {
        printf("Failed to connect, return code %d\n", rc);
        exit(EXIT_FAILURE);
    }
    printf("Subscribing to topic %s\nfor client %s using QoS%d\n\n"
           "Press Q<Enter> to quit\n\n",
           SUB_TOPIC, CLIENTID, QOS);
    MQTTClient_subscribe(client, SUB_TOPIC, QOS);

    // ---------- 3. 共享内存初始设置 ----------

    // 获取共享内存中结构体节点数量
    cJSON *stm32 = cJSON_GetObjectItem(root, "stm32");
    if (stm32 == NULL)
    {
        printf("findstm32 err\n");
        return -1;
    }
    cJSON *modbus = cJSON_GetObjectItem(root, "modbus");
    if (modbus == NULL)
    {
        printf("findmodbus err\n");
        return -1;
    }
    cJSON *stmdata = cJSON_GetObjectItem(stm32, "data");
    if (stmdata == NULL)
    {
        printf("findstmdata err\n");
        return -1;
    }
    cJSON *modbusdata = cJSON_GetObjectItem(modbus, "data");
    if (modbusdata == NULL)
    {
        printf("findmodbusdata err\n");
        return -1;
    }
    stmsize = cJSON_GetArraySize(stmdata);
    int modbussize = cJSON_GetArraySize(modbusdata);
    printf("std_node_num = %d\n", stmsize + modbussize); // 共享内存中数据节点总个数 = stmsize + modbussize

    // 初始化共享内存
    int shmret = -1;
    shmret = shm_init(&para, "shm_test", SHM_SIZE);
    if (shmret < 0)
    {
        return -1;
    }
    void *node_p = shm_getaddr(&para);
    if (node_p == NULL)
    {
        printf("shm_getaddr err\n");
        return -1;
    }
    memset(node_p, 0, SHM_SIZE);
    total = (int *)node_p; // 指向共享内存头
    *total = stmsize + modbussize;
    node_arr = (std_node *)(node_p + sizeof(int)); // 指向共享内存结构体数组头

    // 向共享内存映射结构体节点数据
    for (int i = 0; i < *total; ++i)
    {
        if (i < stmsize)
        {
            cJSON *stmdata_i = cJSON_GetArrayItem(stmdata, i);
            cJSON *key = cJSON_GetObjectItem(stmdata_i, "key");
            cJSON *type = cJSON_GetObjectItem(stmdata_i, "type");
            node_arr[i].key = key->valueint;
            node_arr[i].type = type->valueint;
            node_arr[i].dev_type = 0;
            node_arr[i].ret = 0;
            // printf("%s = %d\n", key->string, node_arr[i].key);
            // printf("%s = %d\n", type->string, node_arr[i].type);
        }
        else
        {
            cJSON *modbusdata_i = cJSON_GetArrayItem(modbusdata, i - stmsize);
            cJSON *key = cJSON_GetObjectItem(modbusdata_i, "key");
            cJSON *type = cJSON_GetObjectItem(modbusdata_i, "type");
            node_arr[i].key = key->valueint;
            node_arr[i].type = type->valueint;
            node_arr[i].dev_type = 1;
            node_arr[i].ret = 0;
            // printf("%s = %d\n", key->string, node_arr[i].key);
            // printf("%s = %d\n", type->string, node_arr[i].type);
        }
    }

    // ---------- 4. 数据上报策略 ----------

    // 解析上报模式，根据模式选择上报线程
    cJSON *report = cJSON_GetObjectItem(root, "report");
    if (report == NULL)
    {
        printf("findreport err\n");
        return -1;
    }
    cJSON *reporttype = cJSON_GetObjectItem(report, "type");
    if (reporttype == NULL)
    {
        printf("findreporttype err\n");
        return -1;
    }
    cJSON *period = cJSON_GetObjectItem(report, "period");
    if (period == NULL)
    {
        printf("findreporttype err\n");
        return -1;
    }

    pthread_t reportid;    // 上报线程pid
    pth_report reportargs; // 存储上报线程变量

    switch (reporttype->valueint)
    {
    case 0:
        // 刷新上报，先传输一次数据，然后等待指令

        reportargs.rpt_id = -1; // 用于在刷新上报线程中区分是第一次自动刷新，还是得到指令刷新
        reportargs.tim = period->valueint;
        if (pthread_create(&reportid, NULL, handler_refresh, &reportargs) != 0)
        {
            perror("pthread_create err");
            return -1;
        }
        break;
    case 1:
        // 变化上报
        // reportargs.rpt_id = reportid; // 多余
        reportargs.tim = period->valueint;
        if (pthread_create(&reportid, NULL, handler_change, &reportargs) != 0)
        {
            perror("pthread_create err");
            return -1;
        }
        break;
    case 2:
        // 周期上报
        // reportargs.rpt_id = reportid; // 多余
        reportargs.tim = period->valueint;
        if (pthread_create(&reportid, NULL, handler_period, &reportargs) != 0)
        {
            perror("pthread_create err");
            return -1;
        }
        break;
    default:
        printf("report_type err\n");
        break;
    }
    pthread_join(reportid, NULL);
    // do
    // {
    //     ch = getchar();
    // } while (ch != 'Q' && ch != 'q');

    MQTTClient_unsubscribe(client, SUB_TOPIC);
    MQTTClient_disconnect(client, 10000);
    MQTTClient_destroy(&client);

    fclose(file);
    cJSON_Delete(root);
    return 0;
}

// ==================== 功能函数实现 ====================

//---------- 上报线程函数 ----------

// 周期上报线程函数
void *handler_period(void *arg)
{
    // 创建JSon，插入数据
    pth_report *reportargs = (pth_report *)arg; // 接收参数
    cJSON *reportdata = cJSON_CreateObject();
    cJSON_AddItemToObject(reportdata, "type", cJSON_CreateNumber(1));   // 回复类型
    cJSON_AddItemToObject(reportdata, "result", cJSON_CreateNumber(0)); // 上报是否成功？
    while (1)
    {
        sleep(reportargs->tim); // 周期
        // 创建数据存放节点
        cJSON *data = cJSON_CreateArray();
        if (data == NULL)
        {
            printf("reporrtdata err\n");
            return NULL;
        }
        cJSON_AddItemToObject(reportdata, "data", data);
        printf("add_report_data begin\n");
        // 遍历插入数据
        for (int i = 0; i < *total; ++i)
        {
            cJSON *partdata = cJSON_CreateObject();
            cJSON_AddItemToObject(partdata, "key", cJSON_CreateNumber(node_arr[i].key));
            char value_str[N]; // 存数据字符串
            // 将数据转为字符串
            switch (node_arr[i].type)
            {
            case 1:
                snprintf(value_str, sizeof(value_str), "%d", node_arr[i].new_val.b_val);
                break;
            case 2:
                snprintf(value_str, sizeof(value_str), "%d", node_arr[i].new_val.i_val);
                break;
            case 3:
                snprintf(value_str, sizeof(value_str), "%f", node_arr[i].new_val.f_val);
                break;
            default:
                printf("data report type err\n");
            }
            // 添加具体数据
            cJSON_AddItemToObject(partdata, "val", cJSON_CreateString(value_str));
            cJSON_AddItemToArray(data, partdata);
            memset(value_str, 0, N);
        }
        char *reportbuf = cJSON_Print(reportdata); // 转字符串
        // mqtt发送
        // printf("%s", reportbuf);

        pubmsg.payload = reportbuf;
        pubmsg.payloadlen = (int)strlen(reportbuf);
        pubmsg.qos = QOS;
        pubmsg.retained = 0;
        MQTTClient_publishMessage(client, PUB_TOPIC, &pubmsg, &token);
        printf("Waiting for up to %d seconds for publication of data\n"
               "on topic %s for client with ClientID: %s\n",
               (int)(TIMEOUT / 1000), PUB_TOPIC, CLIENTID);
        rc = MQTTClient_waitForCompletion(client, token, TIMEOUT);
        printf("Message with delivery token %d delivered\n", token);

        free(reportbuf);
        cJSON_DeleteItemFromObject(reportdata, "data");
    }

    pthread_exit(NULL);
}

// 变化上报线程函数

void *handler_change(void *arg)
{
}

// 刷新上报函数

void *handler_refresh(void *arg)
{
    pth_report *refresh_args = (pth_report *)arg; // 接收参数
    // 创建JSon，插入数据
    cJSON *reportdata = cJSON_CreateObject();                           // 最外层{}
    cJSON_AddItemToObject(reportdata, "type", cJSON_CreateNumber(1));   // 回复类型
    cJSON_AddItemToObject(reportdata, "result", cJSON_CreateNumber(0)); // 上报是否成功？

    // 创建数据存放节点
    cJSON *data = cJSON_CreateArray();
    if (data == NULL)
    {
        printf("reporrtdata err\n");
        return NULL;
    }
    cJSON_AddItemToObject(reportdata, "data", data);
    printf("add_report_data begin\n");

    // 遍历插入数据
    for (int i = 0; i < *total; ++i)
    {
        cJSON *partdata = cJSON_CreateObject();
        cJSON_AddItemToObject(partdata, "key", cJSON_CreateNumber(node_arr[i].key));
        char value_str[N]; // 存数据字符串
        // 将数据转为字符串
        switch (node_arr[i].type)
        {
        case 1:
            snprintf(value_str, sizeof(value_str), "%d", node_arr[i].new_val.b_val);
            break;
        case 2:
            snprintf(value_str, sizeof(value_str), "%d", node_arr[i].new_val.i_val);
            break;
        case 3:
            snprintf(value_str, sizeof(value_str), "%f", node_arr[i].new_val.f_val);
            break;
        default:
            printf("data report type err\n");
        }
        // 添加具体数据
        cJSON_AddItemToObject(partdata, "val", cJSON_CreateString(value_str));
        cJSON_AddItemToArray(data, partdata);
        memset(value_str, 0, N);
    }
    char *reportbuf = cJSON_Print(reportdata); // 转字符串

    // mqtt发送
    pubmsg.payload = reportbuf;
    pubmsg.payloadlen = (int)strlen(reportbuf);
    pubmsg.qos = QOS;
    pubmsg.retained = 0;
    MQTTClient_publishMessage(client, PUB_TOPIC, &pubmsg, &token);
    printf("Waiting for up to %d seconds for publication of data\n"
           "on topic %s for client with ClientID: %s\n",
           (int)(TIMEOUT / 1000), PUB_TOPIC, CLIENTID);
    rc = MQTTClient_waitForCompletion(client, token, TIMEOUT);
    printf("Message with delivery token %d delivered\n", token);

    free(reportbuf);
    cJSON_DeleteItemFromObject(reportdata, "data");

    // 接收指令创建的线程,释放malloc创建的空间
    if (refresh_args->rpt_id != -1)
    {
        free(refresh_args);
    }
    pthread_exit(NULL);
}

//---------- mqtt相关函数 ----------

void *handler_msg(void *arg)
{
    pth_msg_recv *msg_recv_args = (pth_msg_recv *)arg; // 接收参数

    // 解析命令，响应，转发
    cJSON *msg_json = cJSON_Parse(msg_recv_args->msg);       // 将消息转为json格式
    cJSON *msg_type = cJSON_GetObjectItem(msg_json, "type"); // 指令类型
    if (msg_type->valueint == 1)
    {
        // 刷新上报时使用，接收指令刷新数据
        pth_report *refresh_args = (pth_report *)malloc(sizeof(pth_report)); // 防止线程id丢失

        if (pthread_create(&refresh_args->rpt_id, NULL, handler_refresh, refresh_args) != 0)
        {
            perror("pthread_create err");
            return -1;
        }
        pthread_detach(refresh_args->rpt_id); // 将线程设置为分离态，结束自动回收资源
    }
    else if (msg_type->valueint == 2)
    {
        //
    }
    else if (msg_type->valueint == 3)
    {
    }
    else if (msg_type->valueint == 4)
    {
    }
    free(msg_recv_args->msg);
    free(msg_recv_args);
    pthread_exit(NULL);
}

// 消息投递确认回调
void delivered(void *context, MQTTClient_deliveryToken dt)
{
    printf("Message with token value %d delivery confirmed\n", dt);
    deliveredtoken = dt;
}

// MQTT消息到达回调
int msgarrvd(void *context, char *topicName, int topicLen, MQTTClient_message *message)
{
    int i;
    char *payloadptr;
    // 每来一个消息创建一个指令线程去处理，处理完消息直接关闭此线程
    pth_msg_recv *msg_recv_args = (pth_msg_recv *)malloc(sizeof(pth_msg_recv)); // 防止线程id丢失
    strcpy(msg_recv_args->msg, message->payload);                               // 复制消息
    // printf("Message arrived\n");
    // printf("     topic: %s\n", topicName);
    // payloadptr = message->payload;
    // for (i = 0; i < message->payloadlen; i++)
    // {
    //     putchar(*payloadptr++);
    // }
    // putchar('\n');

    if (pthread_create(&msg_recv_args->msg_id, NULL, handler_msg, msg_recv_args) != 0)
    {
        perror("pthread_create err");
        return -1;
    }
    pthread_detach(msg_recv_args->msg_id); // 将线程设置为分离态，结束自动回收资源

    MQTTClient_freeMessage(&message);
    MQTTClient_free(topicName);
    return 1;
}

// 连接丢失回调
void connlost(void *context, char *cause)
{
    printf("\nConnection lost\n");
    printf("     cause: %s\n", cause);
}